package com.yanggu.networkflow_analysis

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.util.Random

object PageView {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //env.setParallelism(1)

    env
      .readTextFile(getClass.getResource("/UserBehavior.csv").getPath)
      //转换成样例类, 并且只关注pv
      .flatMap((dataLine, out: Collector[UserBehavior]) => {
        val arr = dataLine.split(",")
        //这里只关注pv
        if (arr(3) == "pv") {
          out.collect(UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, "pv", arr(4).toLong))
        }
      })
      //设置EventTime, 这里不设置Watermark, 自己决定窗口的数据什么时候进行计算
      .assignAscendingTimestamps(_.timestamp * 1000)
      //将数据映射成随机数, 为了充分利用并行度进行分布式计算, 而不是使用timeWindowAll, 这样数据倾斜太严重
      .map(_ => Random.nextString(8))
      //根据随机key进行分组
      .keyBy(data => data)
      //设置滚动事件时间窗口, 窗口大小1小时
      .window(TumblingEventTimeWindows.of(Time.hours(1L)))
      //执行增量聚合
      .aggregate(new PvCountAgg, new PvCountWindowResult)
      //由于按照随机key进行分组后, 需要将数据重新进行聚合, 按照windowEnd进行分组
      .keyBy(_.windowEnd)
      //注册定时器, 窗口结束后输出数据
      .process(new PvCountKeyedProcessFunction)
      .print("pv count")

    //执行job
    env.execute("PageView Job")
  }

}

/**
 * 用户行为的实体类。输入数据的样例类
 *
 * @param userId     用户id
 * @param itemId     商品id
 * @param categoryId 商品的分类id
 * @param behavior   具体的行为
 * @param timestamp  时间戳。这里是秒，不是毫秒
 */
case class UserBehavior(userId: Long, itemId: Long, categoryId: Int, behavior: String, timestamp: Long)

case class PvCount(windowEnd: Long, count: Long)

class PvCountAgg extends AggregateFunction[String, Long, Long] {

  override def createAccumulator() = 0L

  override def add(value: String, accumulator: Long) = accumulator + 1L

  override def getResult(accumulator: Long) = accumulator

  override def merge(a: Long, b: Long) = a + b

}

class PvCountWindowResult extends ProcessWindowFunction[Long, PvCount, String, TimeWindow] {

  override def process(key: String, context: Context, elements: Iterable[Long], out: Collector[PvCount]): Unit = {
    out.collect(PvCount(context.window.getEnd, elements.head))
  }

}

class PvCountKeyedProcessFunction extends KeyedProcessFunction[Long, PvCount, PvCount] {

  private lazy val valueState = getRuntimeContext.getState(new ValueStateDescriptor[Long]("pv-count", classOf[Long]))

  override def processElement(value: PvCount,
                              ctx: KeyedProcessFunction[Long, PvCount, PvCount]#Context,
                              out: Collector[PvCount]) = {
    valueState.update(valueState.value() + value.count)
    //注册定时器, 在窗口结束时间 + 1ms后执行
    ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey + 1)
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[Long, PvCount, PvCount]#OnTimerContext,
                       out: Collector[PvCount]) = {
    //直接输出即可
    out.collect(PvCount(ctx.getCurrentKey, valueState.value()))
    //清空状态
    valueState.clear()
  }

}
